package com.atguigu.app

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{CouponAlertInfo, EventLog}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import scala.util.control.Breaks._

object AlertApp {
  def main(args: Array[String]): Unit = {
    //1.创建sparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("AlertApp").setMaster("local[*]")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf,Seconds(5))

    //3.消费kafka数据
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT,ssc)

    //4.将数据转为样例类，并补全字段
    val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
    val eventLogDStream = kafkaDStream.map(record => {
      val eventLog: EventLog = JSON.parseObject(record.value(), classOf[EventLog])

      //补全字段
      val times: String = sdf.format(new Date(eventLog.ts))
      eventLog.logDate = times.split(" ")(0)
      eventLog.logHour = times.split(" ")(1)
      (eventLog.mid,eventLog)
    })


    //5.开启一个5min的窗口
    val windowDStream = eventLogDStream.window(Minutes(5))

    //6.对相同mid得到数据聚合到一块
    val midToLogDStream: DStream[(String, Iterable[EventLog])] = windowDStream.groupByKey()

    //7.跟据条件筛选数据 组成疑似预警日志
    /**
      * 三次及以上用不同账号登录并领取优惠劵，（先判断用户是否领优惠券，如果领优惠劵，对领优惠券的用户去重，然后计算用户个数）
      * 并且过程中没有浏览商品。（判断用户是否浏览商品，如果浏览商品，则不符合以上要求，可以直接过滤）
      */
    val boolToInfoDStream: DStream[(Boolean, CouponAlertInfo)] = midToLogDStream.map { case (mid, iter) =>
      //创建用来存放领优惠券的用户id集合
      val uids: util.HashSet[String] = new util.HashSet[String]()
      //创建用来存放领优惠券所涉及的商品ID集合
      val itemIds: util.HashSet[String] = new util.HashSet[String]()
      //创建用户涉及行为的集合
      val events: util.ArrayList[String] = new util.ArrayList[String]()

      //创建一个标志位，用来判断5min内是否有浏览商品行为
      var bool: Boolean = true

      //遍历相同mid 5min的数据
      breakable {
        iter.foreach(log => {
          //可能存在浏览商品行为
          events.add(log.evid)
          //判断用户是否浏览商品
          if ("clickItem".equals(log.evid)) {
            //证明有浏览商品行为，不符合预警需求，则跳出循环
            bool = false
            break()
          } else if ("coupon".equals(log.evid)) {
            //没有浏览商品但是领优惠券
            //可能用户个数不超过三个
            uids.add(log.uid)
            itemIds.add(log.itemid)
          }
        })
      }
      //生成疑似预警日志
      (uids.size() >= 3 && bool, CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis()))
    }
    //8.生成预警日志
    val couponAlertInfoDStream: DStream[CouponAlertInfo] = boolToInfoDStream.filter(_._1).map(_._2)

    couponAlertInfoDStream.print()

    //9.将数据写入ES 并且相同mid每分钟只保留一次
    couponAlertInfoDStream.foreachRDD(rdd=>{
      rdd.foreachPartition(partiton=>{
        val list: List[(String, CouponAlertInfo)] = partiton.toList.map(log => {
          (log.mid + log.ts / 1000 / 60, log)
        })
        MyEsUtil.insertBulk(GmallConstants.ES_ALERT+"220309",list)
      })
    })

    //10.启动任务
    ssc.start()
    ssc.awaitTermination()
  }
}
